package com.atguigu.flink.chapter06.state;

import com.atguigu.flink.function.WaterSensorMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.runtime.state.memory.MemoryStateBackend;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.io.IOException;

/**
 * @Author lzc
 * @Date 2023/6/25 11:24
 */
public class StateBackendDemo {
    public static void main(String[] args) throws IOException {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(2);
    
        
        // 设置状态后端
//        env.setStateBackend(new HashMapStateBackend()); // 设置 hashMap 状态后端
//        env.setStateBackend(new EmbeddedRocksDBStateBackend()); // 设置 rocksDb 状态后端
        env.setStateBackend(new EmbeddedRocksDBStateBackend(true));
        // 远程状态存储(Checkpoint 存储地方)
        //env.getCheckpointConfig().setCheckpointStorage(new JobManagerCheckpointStorage()); // Checkpoint 存储JobManager内存
       // env.getCheckpointConfig().setCheckpointStorage("hdfs:/..."); // Checkpoint 存储到 hdfs
        
        
        // 旧的写法:
        // 1. 本地在TaskManager 内存 2. 远程在JobManager内存
        env.setStateBackend(new MemoryStateBackend());
        
        // 2. 本地在 TaskManager 的内存 2. 远程在 Hdfs
        env.setStateBackend(new FsStateBackend("hdfs:/....."));
        
        // 3. 本地在 RocksDb 2. 远程在 hdfs
        env.setStateBackend(new RocksDBStateBackend("hdfs:/...."));
        
        
        env
            .socketTextStream("hadoop162", 8888)
            .map(new WaterSensorMapFunction())
            .print();
        
        
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    

    
    
}
